#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import locale
import os
import platform
import signal
import time

from common.common import execute_cmd, read_result_file, touch_file
from common.common_models import LogDetail
from common.const import CMDResult, DBLogLevel, SubJobStatusEnum
from common.file_common import delete_file, delete_path, create_dir_recursive
from common.number_const import NumberConst
from oracle import logger
from oracle.common.common import chown_file_path_owner, get_compatible_info, log_func_time_cost, \
    delete_matching_start_with_pattern_files, mount_bind_path, umount_bind_path, exec_sql_cmd, parse_html_result
from oracle.common.common_expansion import report_job_details_linux_windows, delete_matching_contain_pattern_files
from oracle.common.constants import PluginPathConstant, RestoreType, RETURN_INVALID, ScriptExitCode, \
    OracleReportLabel, OracleDataBaseType, RestoreByType, PexpectResult
from oracle.common.copy_func import get_copy_db_info, CopyDBInfo, get_copy_files
from oracle.common.linux_common import execute_linux_rman_cmd
from oracle.common.user_env_common import get_oracle_user_env_by_linux
from oracle.common.windows_common import execute_windows_rman_cmd, redirect_backup_path, \
    write_tmp_file, execute_windows_cmd
from oracle.schemas.oracle_schemas import DBFile
from oracle.services.restore.restore_common import shutdown_abort, login_oracle_database_restore
from oracle.services.restore.restore_func import get_omf_dest, build_omf_recovery_file_dest
from oracle.services.restore.windows_sbin_func import unix_timestamp_to_date


class AssistInsOracleRestore:
    def __init__(self, pid, script_root_path, job_id, sub_job_id, rec_memory_limit_gb, block_change_tracking):
        self.rec_memory_limit_gb = rec_memory_limit_gb
        self.sql_params = {}
        self.rman_params = {}
        self.additional = ''
        self.auth_type = ''
        self.spfile_location_new = ''
        self.spfile_location = ''
        self.task_type = 0
        self.resetlogs_id = ''
        self.pfile_name = ''
        self.dbpw_file = ''
        self.upgrade = ''
        self.incar_num = ''
        self.backup_all_version = ''
        self.backup_pre_version = ''
        self.export_oracle_env = ''
        self.rman_enc_section = ''
        self.is_enc_bk = 0
        self.pit = ''
        self.copy_restore = 0
        self.main_backup_path = ''
        self.backup_path_list = ''
        self._pid = pid
        self._sub_job_id = sub_job_id
        self._job_id = job_id
        self.asm_user = ''
        self.asm_user_pwd = ''
        self.channels = 0
        self.pit_time = ''
        self.pit_scn = ''
        self.recover_target = ''
        self.recover_path = ''
        self.recover_order = ''
        self.storage_type = ''
        self.asm_sid_name = ''
        self.enc_algo = ''
        self.enc_key = ''
        self.pfile_pid = ''
        self.restore_by = ''
        self.recover_num = ''
        self.backup = ''
        self.meta_data_path = ''
        self.archive = ''
        self.job_id_list = ''
        self.bct_status = ""
        self.assist_ins_sid = ''
        self.block_change_tracking = block_change_tracking
        self._init_db_param()
        self.omf_dest = ''
        self.copy_db_files: [DBFile] = []

    def get_restore_vars(self, params, params_pdb):
        self.db_name = params['AppName']
        self.db_instance = params['InstanceName']
        self.original_db_instance = params['OriginalInstanceName']
        self.db_user = params['UserName']
        self.db_user_pwd = params['Password']
        self.oracle_home = params['OracleHome']
        self.oracle_base = params['OracleBase']
        self.access_oracle_home = params['AccessOracleHome']
        self.access_oracle_base = params['AccessOracleBase']
        self.asm_user = params['ASMUserName']
        self.asm_user_pwd = params['ASMPassword']
        self.channels = params['Channel']
        self.pit_time = params['pitTime']
        self.pit_scn = params['pitScn']
        self.recover_target = params['recoverTarget']
        self.recover_path = params['recoverPath']
        self.recover_order = params['recoverOrder']
        self.storage_type = params['storType']
        self.asm_sid_name = params['ASMInstanceName']
        self.db_type = params['dbType']
        self.enc_algo = params['EncAlgo']
        self.enc_key = params['EncKey']
        self.pfile_pid = params['pfilePID']
        self.restore_by = params['RestoreBy']
        self.recover_num = params['recoverNum']
        self.ora_db_user = params['OracleInstallUser']
        self.ora_grid_user = params['GridInstallUser']
        self.backup = params['DataPath']
        self.meta_data_path = params['MetaDataPath']
        self.archive = params['LogPath']
        self.job_id_list = params['JobIdList']
        self.bct_status = params['bctStatus']
        self.target_db_run_user = params['targetDbRunUser']
        self.target_db_run_user_pwd = params['targetDbRunUserPwd']
        self.pdb_names = params_pdb['pdb_names']
        self.is_pdb_restore = params_pdb['is_pdb_restore']
        self.resetlogs_id = params_pdb['resetlogs_id']
        self.rman_params = {
            "is_enc_bk": self.is_enc_bk,
            "instance_name": self.get_assist_ins_sid(),
            "db_user": self.db_user,
            "db_password": self.db_user_pwd,
            "rman_enc_section": "",
            "db_install_user": self.ora_db_user
        }

    def get_init_assist_ins_vars(self, params):
        self.copy_sub_type = params['copySubType']
        self.parent_subtype = params['parent_subtype']

    def define_params(self):
        logger.info(f"Enter function define_params, pid:{self._pid}.")
        self.backup_path_list = redirect_backup_path(self.backup)
        self.main_backup_path = self.backup_path_list.split(";")[0]
        if platform.system().lower() == "windows":
            self.mount_main_backup_path = self.main_backup_path
        else:
            mount_bind_path_prefix = PluginPathConstant.ORACLE_DATA_FILE_PATH.value
            logic_ip_path = self.main_backup_path.split('/')[-1]
            self.mount_main_backup_path = mount_bind_path_prefix + logic_ip_path + self.assist_ins_sid
            create_dir_recursive(self.mount_main_backup_path)
            if not mount_bind_path(self.main_backup_path, self.mount_main_backup_path):
                logger.error(f"Mount bind main path failed, job_id:{self._job_id}.")
                return ScriptExitCode.ERROR_MOUNTPATH
        self.additional = os.path.join(self.main_backup_path, "additional")
        backup_temp = os.path.join(self.main_backup_path, "tmp")
        logger.info(f'BACKUPTEMP={backup_temp};ADDITIONAL={self.additional};'
                    f'MainBackupPath={self.main_backup_path}, pid:{self._pid}.')
        encoding = locale.getdefaultlocale()[1]
        scn_dbf_max_content = read_result_file(os.path.join(self.additional, "scn_dbf_max"), encoding=encoding).strip()
        self.define_params_by_scn_dbf(scn_dbf_max_content)
        pit_time_date = ''
        if self.pit_time:
            pit_time_date = unix_timestamp_to_date(self.pit_time)
            logger.info(f"Convert {self.pit_time} to {pit_time_date}, pid:{self._pid}.")
        if self.pit_scn != "0":
            self.pit = f"scn {self.pit_scn}"
        else:
            self.pit = f"time \"to_date('{pit_time_date}', 'YYYY-MM-DD HH24:MI:SS')\""
        if self.channels:
            self.channels = int(self.channels)
        if not self.channels or self.channels == 0:
            self.channels = 4
            logger.warning(F"Setting channels number to {self.channels} by default, pid:{self._pid}.")
        if self.channels > 254:
            logger.error(f"Channels param is invalid, Channels={self.channels}, pid:{self._pid}.")
            return RETURN_INVALID
        self.is_enc_bk = 0
        logger.info(f"PID={self._pid};DBINSTANCE={self.db_instance};DBNAME={self.db_name};DBUSER={self.db_user};"
                    f"ASMSIDNAME={self.asm_sid_name};ASMUSER={self.asm_user};DBCHANNEL={self.channels};"
                    f"PIT_TIME={self.pit_time};PIT_SCN={self.pit_scn};BACKUP={self.backup};PIT={self.pit}"
                    f"ARCHIVE={self.archive};RECOVERTARGET={self.recover_target};RECOVERPATH={self.recover_path};"
                    f"RECOVERORDER={self.recover_order};pfilePID={self.pfile_pid};RestoreBy={self.restore_by};"
                    f"RECOVERNUM={self.recover_num};JOBID_LIST={self.job_id_list};ORACLE_HOME={self.oracle_home}")
        return ScriptExitCode.SUCCESS

    def define_params_by_scn_dbf(self, scn_dbf_max_content):
        pit_scn_max = scn_dbf_max_content.split()[0]
        logger.info(f"pit_scn_max={pit_scn_max}")
        if self.pit_scn and int(self.pit_scn) < int(pit_scn_max):
            logger.info(
                f"Base media snapshot used to restore is newer than PIT to restore {self.pit_time}, pid:{self._pid}.")
        if not self.pit_time and not self.pit_scn:
            self.pit_scn = pit_scn_max
        # 0 is arch resestore 1 is full restore
        self.copy_restore = RestoreType.ARCHICE_RESTORE
        if not self.archive:
            logger.info(f"ARCHIVE is null, restore by full copy, pid:{self._pid}.")
            self.copy_restore = RestoreType.FULL_RESTORE
        if self.restore_by == RestoreType.FULL_RESTORE:
            logger.info(f"Full backup, restore by copy, pid:{self._pid}.")
            self.copy_restore = RestoreType.FULL_RESTORE

    @log_func_time_cost
    def mount_assist_ins(self):
        logger.info(f"start mount assist database, pid: {self._pid}.")
        ret_code = self.define_params()
        if ret_code != ScriptExitCode.SUCCESS:
            logger.error(f"Exec define_params error, ret: {ret_code}")
            return ret_code

        # 创建pfile
        ret_code = self.create_pfile()
        if ret_code != ScriptExitCode.SUCCESS:
            logger.error(f"Exec create_pfile error, ret: {ret_code}")
            return ret_code
        # 数据库认证需要创建密码文件
        if self.db_user_pwd and not self.create_pwd_file():
            logger.error(f"Exec create_pfile error, ret: {ret_code}")
            return RETURN_INVALID

        # rman恢复
        logger.info(f"start recover assist database with pfile: {self.get_assist_ins_pfile_path()}.")
        # 构造rman语句
        sql_content = self.build_restore_ctrl_sql_content()
        ret_code = self.execute_rman_cmd(sql_content)
        if ret_code != ScriptExitCode.SUCCESS:
            logger.error(f"Exec execute_rman_cmd  error, ret: {ret_code}")
            return ret_code

        sql_content = self.build_rman_sql_content()
        ret_code = self.execute_rman_cmd(sql_content)
        if ret_code != ScriptExitCode.SUCCESS:
            logger.error(f"Exec execute_rman_cmd error, ret: {ret_code}")
            return ret_code
        logger.info(f"end mount assist database, pid: {self._pid}, sid: {self.get_assist_ins_sid()}.")
        return ScriptExitCode.SUCCESS

    def build_restore_ctrl_sql_content(self):
        sql_content = (f"startup nomount pfile='{self.get_assist_ins_pfile_path()}';\n"
                       f"restore controlfile from '{os.path.join(self.mount_main_backup_path, 'controlfile.ctl')}';\n"
                       f"startup mount;\n"
                       f"CONFIGURE DEVICE TYPE DISK PARALLELISM {self.channels};\n"
                       f"crosscheck datafilecopy all;\n"
                       f"delete noprompt expired datafilecopy all;\n"
                       f"delete noprompt expired backupset;\n"
                       f"crosscheck archivelog all;\n"
                       "delete noprompt expired archivelog all;\n"
                       "quit;")
        return sql_content

    def build_rman_sql_content(self):
        # 获取pdb集恢复需要跳过的表空间
        skip_str = ""
        if self.is_pdb_restore:
            skip_str = self.get_skip_pdb_tablespace()
        copy_db_info: CopyDBInfo = get_copy_db_info(self.main_backup_path)
        sql_content = (f"RESET DATABASE TO INCARNATION {copy_db_info.incarnation_number};\n"
                       f"RUN\n"
                       f"{{\n"
                       f"    SET COMMAND ID TO 'ProtectAgent_Restore';\n"
                       f"{self.build_channel_and_parallelism_sql()}"
                       f"{self.build_catalog_sql()}"
                       f"{self.build_datafile_sql()}"
                       f"{self.build_online_log_sql()}")
        if self.is_pdb_restore and self.restore_by == RestoreByType.TIME:
            sql_content += f"    SET UNTIL {self.get_pit_time()};\n"
        else:
            sql_content += f"    SET UNTIL scn {self.get_scn()};\n"
        sql_content += f"    SWITCH DATAFILE ALL;\n"
        if self.is_pdb_restore and skip_str:
            sql_content += (f"RECOVER DATABASE skip forever tablespace\n"
                            f"{skip_str};\n")
        else:
            sql_content += f"    RECOVER DATABASE;\n"
        sql_content += f"{self.build_tempfile_sql()}\n"
        sql_content += (f"{self.build_release_channel_sql()}"
                        f"}}\n")
        if (self.copy_sub_type == OracleDataBaseType.ORACLE_CLUSTER or
                (self.is_pdb_restore and self.parent_subtype == OracleDataBaseType.ORACLE_CLUSTER)):
            sql_content += f"ALTER DATABASE SET STANDBY TO MAXIMIZE PERFORMANCE;\n"
        if self.block_change_tracking:
            sql_content += f"ALTER DATABASE DISABLE BLOCK CHANGE TRACKING\n;"
        sql_content += f"alter database open RESETLOGS;\n"
        return sql_content

    @log_func_time_cost
    def create_pfile(self):
        logger.info(f"start create assist database pfile, _pid: {self._pid}.")

        # 读取原pfile信息
        ret_code, rst = self.read_pfile_from_backup()
        if ret_code != ScriptExitCode.SUCCESS:
            logger.error(f"Exec read_pfile_from_backup error, ret: {ret_code}")
            return ret_code

        if platform.system().lower() == "windows":
            diagnostic_dest = f"*.diagnostic_dest='{self.oracle_home}'\n"
        else:
            diagnostic_dest = f"*.diagnostic_dest='{self.mount_main_backup_path}'\n"
        # 读原始pfile信息，修改
        new_pfile_data = (f"*.compatible={rst.get('compatible')}\n"
                          f"*.db_name={rst.get('db_name')}\n"
                          f"*.db_files={rst.get('db_files')}\n"
                          f"*.db_unique_name='{self.get_assist_ins_sid()}'\n"
                          f"*.sga_target={str(int(int(self.rec_memory_limit_gb) * 1024 * 0.75))}m\n"
                          f"*.control_files='{os.path.join(self.mount_main_backup_path, 'controlfile.ctl')}'\n"
                          f"*.pga_aggregate_target={str(int(int(self.rec_memory_limit_gb) * 1024 * 0.25))}m\n"
                          f"{diagnostic_dest}"
                          f"*.audit_file_dest='{self.mount_main_backup_path}'\n")

        if rst.get('is_cdb'):
            new_pfile_data += f"*.enable_pluggable_database=true\n"
        backup_pfile = os.path.join(self.mount_main_backup_path, f"ebackup-{self.db_name}-pfile.ora")
        omf_dest = get_omf_dest(backup_pfile)
        if omf_dest:
            self.omf_dest = omf_dest
            omf_dict = build_omf_recovery_file_dest(self.db_name, self.ora_db_user, self.mount_main_backup_path)
            new_pfile_data += (f"*.control_files='{omf_dict.get('control_files')}'\n"
                               f"*.db_create_file_dest='{omf_dict.get('db_create_file_dest')}'\n")
        else:
            new_pfile_data += f"*.control_files='{os.path.join(self.mount_main_backup_path, 'controlfile.ctl')}'\n"
            pass

        # 写文件
        with open(self.get_assist_ins_pfile_path(), mode='w', encoding='utf-8') as file:
            file.write(new_pfile_data)
        if platform.system().lower() != "windows":
            chown_file_path_owner(self.get_assist_ins_pfile_path(), self.ora_db_user)
        logger.info(f"new pfile content: {new_pfile_data}")
        return ScriptExitCode.SUCCESS

    @log_func_time_cost
    def read_pfile_from_backup(self):
        backup_pfile = os.path.join(self.mount_main_backup_path, f"ebackup-{self.db_name}-pfile.ora")
        db_name = None
        compatible = None
        is_cdb = False
        # oracle中默认值为200
        db_files = NumberConst.HUNDRED * 2
        with open(backup_pfile, 'r') as file:
            for line in file:
                logger.info(f"read line: {line}")
                if ".db_name" in line:
                    db_name = line.split("=", 1)[1]
                elif ".compatible" in line:
                    compatible = line.split("=", 1)[1]
                elif ".enable_pluggable_database" in line:
                    is_cdb = ("true" in line.split("=", 1)[1].lower())
                elif ".db_files" in line:
                    db_files = line.split("=", 1)[1]
                else:
                    logger.info(f"Skipping line: {line}")
        if db_name is None or compatible is None or is_cdb is None:
            logger.error(f"cannot read params from pfile, "
                         f"db_name: {db_name}, "
                         f"compatible: {compatible}, "
                         f"is_cdb: {is_cdb}")
            return ScriptExitCode.ERROR_SCRIPT_EXEC_FAILED, None
        return ScriptExitCode.SUCCESS, {
            "db_name": db_name,
            "db_files": db_files,
            "compatible": compatible,
            "is_cdb": is_cdb
        }

    def get_assist_ins_pfile_path(self):
        return os.path.join(self.oracle_home, 'dbs', f"init{self.get_assist_ins_sid()}.ora")

    def remove_assist_definite_file(self, assist_ins_sid):
        ret_code = ScriptExitCode.SUCCESS
        dbs_file_list = [
            os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_DBS_PATH, f"init{assist_ins_sid}.ora"),
            os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_DBS_PATH, f"orapw{assist_ins_sid}"),
            os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_DBS_PATH, f"hc_{assist_ins_sid}.dat"),
            os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_DBS_PATH, f"lkHW{assist_ins_sid[2:]}"),
            os.path.join(self.oracle_home, PluginPathConstant.ORACLE_BASE_ADMIN_PATH, assist_ins_sid),
            os.path.join(self.oracle_base, PluginPathConstant.ORACLE_BASE_ADMIN_PATH, assist_ins_sid),
            os.path.join(self.oracle_home, f'table_dump_{self._job_id}'),
            os.path.join(PluginPathConstant.WINDOWS_TMP_PATH, f"{assist_ins_sid}_expdp.par"),
            os.path.join(PluginPathConstant.WINDOWS_TMP_PATH, f"{assist_ins_sid}_impdp.par"),
            os.path.join(self.oracle_home, "database", f"PWD{assist_ins_sid}.ora"),
            os.path.join(self.oracle_home, "database", f"hc_{assist_ins_sid}.dat")
        ]
        try:
            for dbs_file in dbs_file_list:
                delete_path(dbs_file)
                logger.info(f"Deleted: {dbs_file}")
        except Exception as e:
            logger.error(f"remove assist definite file fail, error is {e}")
            ret_code = ScriptExitCode.ERROR_SCRIPT_EXEC_FAILED
        return ret_code

    def remove_assist_log_and_audit_file(self, assist_ins_sid):
        ret_code = ScriptExitCode.SUCCESS
        log_file_path = os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_LOG_PATH)
        audit_file_path = os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_AUDIT_PATH)
        trace_file_path = os.path.join(self.oracle_home, PluginPathConstant.ORACLE_HOME_RDBMS_PATH,
                                       PluginPathConstant.ORACLE_HOME_TRACE_PATH)
        try:
            delete_matching_start_with_pattern_files(log_file_path, assist_ins_sid)
            delete_matching_start_with_pattern_files(audit_file_path, assist_ins_sid)
            if platform.system().lower() == "windows":
                delete_matching_start_with_pattern_files(trace_file_path, assist_ins_sid)
                delete_matching_contain_pattern_files(self.oracle_base, assist_ins_sid)
        except Exception as e:
            logger.error(f"remove assist log file or audit file fail, error is {e}")
            ret_code = ScriptExitCode.ERROR_SCRIPT_EXEC_FAILED
        return ret_code

    def shutdown_assist_ins(self):
        db_login_params = {
            'pid': self._pid,
            "instance_name": self.assist_ins_sid,
            "db_install_user": self.ora_db_user,
            'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
        }
        shutdown_abort(db_login_params)

    def clear_assist_ins(self, pdb):
        # 清理参数文件
        assist_ins_sid = self.get_assist_ins_sid()
        windows_clear_ret = True
        logger.info(f"Start cleaning up assist instance resources, assist_ins_sid is {assist_ins_sid}, pdb is {pdb}")
        if not pdb:
            self.shutdown_assist_ins()
        if platform.system().lower() == "windows":
            windows_clear_ret, result = execute_windows_cmd(self._pid,
                                                            f"oradim -delete  -sid {self.get_assist_ins_sid()}".split(),
                                                            "")
            if not windows_clear_ret:
                logger.error(f"fail to execute 'oradim -delete -sid {self.get_assist_ins_sid()}', result is {result}")
        rm_definite_ret_code = self.remove_assist_definite_file(assist_ins_sid)
        rm_log_and_audit_ret_code = self.remove_assist_log_and_audit_file(assist_ins_sid)
        if platform.system().lower() != "windows":
            mount_bind_path_prefix = PluginPathConstant.ORACLE_DATA_FILE_PATH.value
            logic_ip_path = self.main_backup_path.split('/')[-1]
            mount_main_backup_path = mount_bind_path_prefix + logic_ip_path + self.assist_ins_sid
            if not umount_bind_path(mount_main_backup_path):
                logger.error(f"Umount bind main path failed, job_id:{self._job_id}.")
        if (rm_definite_ret_code != ScriptExitCode.SUCCESS or rm_log_and_audit_ret_code != ScriptExitCode.SUCCESS
                or (not windows_clear_ret)):
            logger.error(f'fail to clear sub ins resources, pid is {self._pid}')
            log_detail = LogDetail(logInfo=OracleReportLabel.ORACLE_RESTORE_ASSIST_INSTANCE_RESOURCE_RESIDUE_LABEL,
                                   logInfoParam=[assist_ins_sid],
                                   logLevel=DBLogLevel.WARN)
            report_job_details_linux_windows(self._pid, self._job_id, self._sub_job_id, log_detail,
                                             SubJobStatusEnum.RUNNING.value)
        return rm_definite_ret_code + rm_log_and_audit_ret_code

    @log_func_time_cost
    def execute_rman_cmd(self, sql_content):
        tmp_sql = os.path.join(PluginPathConstant.WINDOWS_TMP_PATH, f"tmp_rman_{self._pid}.sql")
        tmp_rst = os.path.join(PluginPathConstant.WINDOWS_TMP_PATH, f"tmp_rman_{self._pid}.rst")
        if platform.system().lower() != "windows":
            tmp_sql = os.path.join(PluginPathConstant.STMP_PATH, f"tmp_rman_{self._pid}.sql")
            tmp_rst = os.path.join(PluginPathConstant.STMP_PATH, f"tmp_rman_{self._pid}.rst")
            touch_file(tmp_sql)
            touch_file(tmp_rst)
            execute_cmd(f"chmod 700 {tmp_sql}")
            execute_cmd(f"chmod 700 {tmp_rst}")
            chown_file_path_owner(tmp_sql, self.ora_db_user)
            chown_file_path_owner(tmp_rst, self.ora_db_user)
        write_tmp_file(tmp_sql, sql_content)
        enc_type = 0
        try:
            if platform.system().lower() == "windows":
                logger.info("execute_windows_rman_cmd begin")
                ret, result = execute_windows_cmd(self._pid, f"oradim -new -sid {self.get_assist_ins_sid()}".split(),
                                                  self.target_db_run_user_pwd)
                if ret:
                    ret_code = execute_windows_rman_cmd(self._pid, self.rman_params, enc_type, tmp_sql, tmp_rst)
                else:
                    ret_code = ScriptExitCode.ERROR_SCRIPT_EXEC_FAILED
                    logger.info(f"fail to execute 'oradim -new -sid {self.get_assist_ins_sid()}', result is {result}")
            else:
                ret_code = execute_linux_rman_cmd(self._pid, self.rman_params, enc_type, tmp_sql, tmp_rst)
            if ret_code != ScriptExitCode.SUCCESS:
                if platform.system().lower() == "windows":
                    encoding = locale.getdefaultlocale()[1]
                    logger.error(f"execute rman cmd: {sql_content} failed, pid:{self._pid}, ret_code:{ret_code}, "
                                 f"error is {read_result_file(tmp_rst, encoding=encoding)}")
                else:
                    logger.error(f"execute rman cmd: {sql_content} failed, pid:{self._pid}, ret_code:{ret_code}, "
                                 f"error is {read_result_file(tmp_rst)}")
                return ret_code
            logger.info(f"execute rman result is: {read_result_file(tmp_rst)}, pid:{self._pid}.")
            logger.info(f"execute rman cmd: {sql_content} success, pid:{self._pid}.")
            return ScriptExitCode.SUCCESS
        except Exception as exception:
            logger.error(f"execute_windows_rman_cmd fail, sql_content={sql_content}, exception={exception},"
                         f" pid={self._pid}.")
            return ScriptExitCode.ERROR_SCRIPT_EXEC_FAILED
        finally:
            delete_file(tmp_sql)
            delete_file(tmp_rst)

    def get_skip_pdb_tablespace(self):
        encoding = locale.getdefaultlocale()[1]
        skip_content = read_result_file(os.path.join(self.additional, "skip_pdb_tablespace"),
                                        encoding=encoding).split()
        skip_str = ""
        if len(skip_content) != 0:
            skip_str = ','.join(str(skip) for skip in skip_content)
        logger.info(f"skip_str={skip_str}")
        return skip_str

    def build_channel_and_parallelism_sql(self):
        tmp_sql_content = ''
        index = 1
        while index <= self.channels:
            formatted_index = '{:02d}'.format(index)
            tmp_sql_content += f"    allocate channel eTableRestore{formatted_index} type disk;\n"
            index += 1
        tmp_sql_content += f"    configure device type disk parallelism {self.channels};\n"
        return tmp_sql_content

    def build_catalog_sql(self):
        sql = f"    catalog start with '{os.path.join(self.mount_main_backup_path)}' noprompt;\n"
        db_file_list = get_copy_files(self.main_backup_path)
        self.copy_db_files = db_file_list
        for db_file in db_file_list:
            parent_dir = str(os.path.join(self.mount_main_backup_path, db_file.file_no))
            for item in os.listdir(str(parent_dir)):
                sql += f"    catalog datafilecopy '{os.path.join(parent_dir, item)}';\n"
        # 适配PDB集日志副本恢复
        if self.is_pdb_restore and self.restore_by == RestoreByType.TIME:
            jobid_list_arr = self.job_id_list.split(",")
            for job_id in jobid_list_arr:
                if not job_id:
                    continue
                sql += f"    catalog start with '{self.archive}/{job_id}/resetlogs_id{self.resetlogs_id}' noprompt;\n"
        return sql

    def build_release_channel_sql(self):
        tmp_sql_content = ''
        index = 1
        while index <= self.channels:
            formatted_index = '{:02d}'.format(index)
            tmp_sql_content += f"    release channel eTableRestore{formatted_index};\n"
            index += 1
        return tmp_sql_content

    def build_datafile_sql(self):
        sql = ""
        for db_file in self.copy_db_files:
            tablespace_name = db_file.ts_name
            file_id = db_file.file_no
            copy_dbf = os.path.join(self.mount_main_backup_path, file_id, f"FNO-{file_id}_TS-{tablespace_name}.dbf")
            sql += f"    SET NEWNAME FOR DATAFILE {file_id} to '{copy_dbf}';\n"
        return sql

    def build_tempfile_sql(self):
        sql = ""
        copy_db_info: CopyDBInfo = get_copy_db_info(self.mount_main_backup_path)
        temp_dbf_list = self.get_tempfile_list()
        if not temp_dbf_list:
            return ""
        time_stamp = int(time.time())
        for key, line in enumerate(temp_dbf_list):
            if not self.omf_dest:
                dst_path = os.path.join(self.mount_main_backup_path, f"tempfile_{key}_{time_stamp}.dbf")
            else:
                dst_path = os.path.join(self.mount_main_backup_path, self.db_name.upper(), "tempfile",
                                        f"tempfile_{key}_{time_stamp}.dbf")
            if int(copy_db_info.ora_pre_version_extend) > 11:
                sql += f"    ALTER DATABASE RENAME FILE '{line.get('NAME')}' to '{dst_path}';\n"
            else:
                sql += f"    SET NEWNAME FOR TEMPFILE {line.get('FILE#')} to '{dst_path}';\n"
        if int(copy_db_info.ora_pre_version_extend) <= 11:
            sql += " SWITCH TEMPFILE ALL;\n"
        return sql

    def get_tempfile_list(self):
        tempfile_dbf_path_list = []
        assist_ins_login_params = {
            'auth_pwd': f"job_copies_0_protectObject_auth_authPwd_{self._pid}",
            'auth_key': f"job_copies_0_protectObject_auth_authKey_{self._pid}"
        }
        result, child = login_oracle_database_restore(
            self._pid, self.get_assist_ins_sid(), assist_ins_login_params, self.ora_db_user)
        if not result:
            logger.error(f'login error, res: {result}.')
            return tempfile_dbf_path_list
        logger.info(f"query assist tempfile:{self.get_assist_ins_sid()}")
        sql = f"set linesize 999;\n select FILE#, NAME from v$tempfile order by FILE#;"
        try:
            return_code, std_out = exec_sql_cmd(child, sql, PexpectResult.FILE_PATH)
            logger.info(f"return_code:{return_code}, std_out:{return_code}")
        except Exception as exception:
            logger.error(f'exec cmd {sql} failed. {str(exception)}')
            return tempfile_dbf_path_list
        finally:
            if platform.system().lower() == "windows":
                child.kill(signal.SIGTERM)
            else:
                child.close()
        if not return_code or PexpectResult.FILE_PATH[3] in str(std_out):
            logger.error(f'select datafile error, return_code: {return_code}. sql: {sql}, stdout:{std_out}')
            return tempfile_dbf_path_list
        dbf_list = parse_html_result(std_out)
        return dbf_list

    def build_online_log_sql(self):
        keep = self.get_keep_identity()
        sql = ""
        encoding = locale.getdefaultlocale()[1]
        logfiles_lines = read_result_file(os.path.join(self.additional, "logfiles"), encoding=encoding).splitlines()
        time_stamp = int(time.time())
        log_num = 0
        for line in logfiles_lines:
            log_path = line.split(";")[1]
            if not self.omf_dest:
                new_log_path = os.path.join(self.mount_main_backup_path, f"restore-redo-{log_num}-{time_stamp}")
            else:
                new_log_path = os.path.join(self.mount_main_backup_path, self.db_name.upper(), "onlinelog",
                                            f"restore-redo-{log_num}-{time_stamp}")
            sql += f"    ALTER DATABASE RENAME FILE '{log_path}' to '{new_log_path}' {keep};\n"
            log_num += 1
        return sql

    def get_keep_identity(self):
        # 对于19版本，要加个keep，防止删除原位置日志文件
        login_params = {
            'pid': self._pid,
            "instance_name": self.db_instance,
            "db_install_user": self.ora_db_user,
            "auth_pwd": f"job_targetObject_auth_authPwd_{self._pid}",
            "auth_key": f"job_targetObject_auth_authKey_{self._pid}"
        }
        compatible = get_compatible_info(self._pid, login_params)
        logger.info(f"compatible: {compatible}")
        compatible_versions = compatible.split(".")
        if int(compatible_versions[0]) >= 19:
            return " KEEP"
        return ""

    def get_scn(self):
        encoding = locale.getdefaultlocale()[1]
        scn_dbf_max_content = read_result_file(os.path.join(self.additional, "scn_dbf_max"), encoding=encoding).strip()
        return scn_dbf_max_content.split()[0]

    def get_pit_time(self):
        pit = ''
        # PDB日志副本恢复，将时间戳转为日期时间格式
        if self.pit_time:
            pit_time_date = unix_timestamp_to_date(self.pit_time)
            logger.info(f"convert_pit_time:{pit_time_date}")
            pit = f"time \"to_date('{pit_time_date}', 'YYYY-MM-DD HH24:MI:SS')\""
        return pit

    def get_assist_ins_sid(self):
        if not self.assist_ins_sid:
            self.assist_ins_sid = f"hw{int(time.time())}"
        return self.assist_ins_sid

    def create_pwd_file(self):
        if platform.system().lower() == "windows":
            dbpw_file_path = os.path.join(self.additional, "dbs", f"PWD{self.db_name}.ora")
            if not os.path.exists(dbpw_file_path):
                return True
            dst_pwd_path = os.path.join(self.oracle_home, "database", f"PWD{self.get_assist_ins_sid()}.ora")
            os.environ["ORACLE_SID"] = self.assist_ins_sid
            os.environ["NLS_LANG"] = "American_America.US7ASCII"
            ret, result = execute_windows_cmd(self._pid,
                                              f"orapwd input_file={dbpw_file_path} file={dst_pwd_path}".split(),
                                              "")
            if not ret:
                logger.error(f"windows create_pwd_file failed, result is {result}")
                return False
        else:
            dbpw_file_path = os.path.join(self.additional, f"dbs/orapw{self.db_name}")
            if not os.path.exists(dbpw_file_path):
                return True
            dst_pwd_path = os.path.join(self.oracle_home, f"dbs/orapw{self.get_assist_ins_sid()}")
            chown_file_path_owner(self.additional, self.ora_db_user)
            chown_file_path_owner(dbpw_file_path, self.ora_db_user)
            oracle_user_env = get_oracle_user_env_by_linux(self.assist_ins_sid)
            create_pwd_file_cmd = f"su - {self.ora_db_user} -c '{oracle_user_env}" \
                                  f" orapwd input_file={dbpw_file_path} file={dst_pwd_path}'"
            return_code, std_out, std_err = execute_cmd(create_pwd_file_cmd)
            if return_code != CMDResult.SUCCESS:
                logger.error(f"create_pwd_file failed, std_out {std_out}, std_err {std_err}")
                return False
        return True

    def _init_db_param(self):
        self.mount_main_backup_path = ''
        self.pdb_names = []
        self.is_pdb_restore = False
        self.db_type = ''
        self.db_name = ''
        self.db_instance = ''
        self.db_user = ''
        self.db_user_pwd = ''
        self.db_is_cluster = 0
        self.target_db_run_user = ''
        self.target_db_run_user_pwd = ''
        self.database_role = ''
        self.ora_version = ''
        self.oracle_is_cdb = ''
        self.ora_db_user = ''
        self.ora_grid_user = ''
        self.oracle_base = ''
        self.ora_db_group = ''
        self.ora_preversion_extend = ''
        self.ora_preversion = ''
        self.oracle_home = ''
        self.original_db_instance = ''
        self.access_oracle_home = ''
        self.access_oracle_base = ''
        self.copy_sub_type = ''
        self.parent_subtype = ''
